/**
 * AggPassiveThread2.java
 * ************************** 
 * @date Apr 6, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.aggregation;

import gossipServices.aggregation.functions.AggregationUpdateFunction;
import gossipServices.basic.nodeDescriptors.ConcreteNodeDescriptor;
import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.basic.view.PartialView;
import gossipServices.gossipMessages.AggregationMessage;
import gossipServices.gossipThread.PassiveGossipThreadInterface;
import gossipServices.network.NetworkServicesAPI;
import gossipServices.pss.PeerSamplingServiceAPI;

import java.net.InetAddress;
import java.net.UnknownHostException;

import messages.Message;

/**
 * Description: 
 * This class contains the instructions of the PassiveThread
 * of the Aggregation protocol. 
 * It extends the AggregationMiddleware class and implements
 * the PassiveGossipThreadInterface. 
 * The implementation of the method
 *  gossipThreadInstructions(), inherited from the base class,
 * calls the passiveThreadInstructions function.
 */
public class AggPassiveThread extends AggregationMiddleware implements
		PassiveGossipThreadInterface {

	/**
	 * @param name
	 * @param cycleLenghtMsec
	 * @param view
	 * @param pss
	 * @param nia
	 * @param function
	 * @param aggStartingValue 
	 */
	public AggPassiveThread(String name, long cycleLenghtMsec,
			PartialView view, PeerSamplingServiceAPI pss,
			NetworkServicesAPI nia, AggregationUpdateFunction function, double startingValue) {
		super(name, cycleLenghtMsec, view, pss, nia, function, startingValue);
	}

	@Override
	public void passiveThreadInstructions() throws InterruptedException {
		AggregationMessage msgQ = (AggregationMessage) nia.receive();
//		log.finer("received: "+msgQ);
		
		NodeDescriptor me = null;
		try {
			me = new ConcreteNodeDescriptor(nia.getNodeName(),
					InetAddress.getLocalHost(), new Integer(0));
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		
		switch(epochHandler.checkEpochs(msgQ, epochHandler)){
		case exchange:
//			log.info(currentThread().getName()+
//					" Epoch: "+epochHandler.getEpoch()
//					+" cycle: "+ epochHandler.getElapsedCycles()+
//					" exchanging with "+msgQ.getSender()+" before exchange: "+estimationValue);
			double sq = msgQ.getSingleValue().getValue();
			Message msg;
			synchronized (estimationValue) {
				msg = new AggregationMessage(me, msgQ.getSender(), estimationValue, 
						AggPassiveThread.class, epochHandler.getEpoch(), epochHandler.getRemainingCycles());			
			}
			nia.send(msg, msgQ.getSender());
//			log.finer("sending: "+((AggregationMessage)msg)+" to "+ msgQ.getSender());
			exchange(sq);
			break;

		case sendCorrectEpoch:
			Message corrMsg = epochHandler.composeCorrectionMessage(me, msgQ.getSender(),
					AggPassiveThread.class); 
//			log.finer("sending correction message "+ corrMsg + " to " + msgQ.getSender() );
			nia.send(corrMsg, msgQ.getSender());
			break;

		case updateCurrentEpoch:
			try {
//				log.finer("updating current epoch from "+msgQ);
				epochHandler.updateCurrentEpoch(msgQ);
			} catch (AggregationRestartException e) {
//				e.printStackTrace();
				restart(e);
			}
			break;
		
		default: 
				
		}
	}

	@Override
	public void gossipThreadInstructions() throws InterruptedException {
		passiveThreadInstructions();
	}

}
